/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal/allocator.h"
#include "dal/subscriber.h"
#include "dal/config_loader.h"
#include "dal_fastdds/converter/StatusConverter.hpp"
#include "dal_fastdds/converter/StatusMaskConverter.hpp"
#include "dal_fastdds/converter/RetTypeConverter.hpp"
#include "dal_fastdds/converter/QosConverter.hpp"
#include "dal_fastdds/converter/RetTypeConverter.hpp"
#include "dal_fastdds/converter/SampleInfoConverter.hpp"
#include "dal_fastdds/converter/GuidConverter.hpp"
#include "dal_fastdds/listener_impl/SubscriberListenerImpl.hpp"
#include "dal_fastdds/listener_impl/DataReaderListenerImpl.hpp"
#include "dal_fastdds/utils/log.hpp"
#include "dal_fastdds/utils/EntityGetter.hpp"
#include "dal_fastdds/utils/EntityMappingManager.hpp"
#include "dal_fastdds/config_manager/ConfigManager.hpp"
#ifdef DAL_ENABLE_DMABUFHEAP
#include "dal_fastdds/dbh/DalDbhBestEffortReader.hpp"
#include "dal_fastdds/dbh/DalDbhReliableReader.hpp"
#include "dal_dmabufheap/DmaBufHeapTopicSelector.hpp"
#endif
#include "dal_fastrtps_shared_cpp/TypeSupport.hpp"
#include "dalidl_typesupport_fastrtps_cpp/message_type_support.h"

using namespace eprosima::fastdds::dds;

// SUBSCRIBER
dal_subscriber_listener_t*
dal_create_subscriber_listener(dal_subscriber_listener_callbacks_t callbacks)
{
    dal_subscriber_listener_t* listener = dal_subscriber_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = new SubscriberListenerImpl(callbacks);
    if (!listener->handle_) {
        LOG_ERROR() << "SubscriberListenerImpl memory allocation failed." << LOG_ENDL;
        dal_subscriber_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_subscriber_listener(dal_subscriber_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    delete reinterpret_cast<SubscriberListenerImpl*>(listener->handle_);
    dal_subscriber_listener_free(listener);
    reinterpret_cast<dal_subscriber_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

dal_subscriber_t* dal_domain_participant_create_subscriber(dal_domain_participant_t* participant,
                                                           dal_subscriber_qos_t* qos,
                                                           dal_subscriber_listener_t* listener,
                                                           dal_status_mask_t mask)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    dal_subscriber_t* d_sub = dal_subscriber_allocate();
    if (!d_sub) {
        LOG_ERROR_WITH_LOCATION() << "dal subscriber allocate failed." << LOG_ENDL;
        return nullptr;
    }

    SubscriberQos f_qos;
    QosConverter::GetInstance()->ConvertSubscriberQosFromDalToFastdds(qos, &f_qos);

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    SubscriberListenerImpl* f_listener =
        listener ? reinterpret_cast<SubscriberListenerImpl*>(listener->handle_) : nullptr;

    auto f_sub = f_participant->create_subscriber(f_qos, f_listener, f_mask);
    if (!f_sub) {
        LOG_ERROR_WITH_LOCATION() << "create subscriber failed." << LOG_ENDL;
        dal_subscriber_free(d_sub);
        return nullptr;
    }

    EntityMappingAddSubscriber(f_sub, d_sub);

    DalSubscriberImpl* handle_ = new DalSubscriberImpl(f_sub, f_listener);
    handle_->SetFastddsDomainParticipant(f_participant);
    d_sub->handle_ = handle_;
    return d_sub;
}

dal_subscriber_t*
dal_domain_participant_create_subscriber_with_profile(dal_domain_participant_t* participant,
                                                      dal_subscriber_qos_t* qos,
                                                      const char* profile_name,
                                                      dal_subscriber_listener_t* listener,
                                                      dal_status_mask_t mask)
{
    if (!profile_name || strlen(profile_name) == 0) {
        return dal_domain_participant_create_subscriber(participant, qos, listener, mask);
    }

    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    dal_subscriber_t* d_sub = dal_subscriber_allocate();
    if (!d_sub) {
        LOG_ERROR_WITH_LOCATION() << "dal subscriber allocate failed." << LOG_ENDL;
        return nullptr;
    }

    SubscriberQos f_qos;
    if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
        try {
            f_participant->get_subscriber_qos_from_profile(profile_name, f_qos);
        }
        catch (...) {
            LOG_ERROR_WITH_LOCATION()
                << "subscriber qos profile[" << profile_name << "] in xml are not effective.";
            dal_subscriber_free(d_sub);
            return nullptr;
        }
    } else {
        QosConverter::GetInstance()->ConvertSubscriberQosFromDalToFastdds(qos, &f_qos);
    }

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    SubscriberListenerImpl* f_listener =
        listener ? reinterpret_cast<SubscriberListenerImpl*>(listener->handle_) : nullptr;

    auto f_sub = f_participant->create_subscriber(f_qos, f_listener, f_mask);
    if (!f_sub) {
        LOG_ERROR_WITH_LOCATION() << "create subscriber failed." << LOG_ENDL;
        dal_subscriber_free(d_sub);
        return nullptr;
    }

    EntityMappingAddSubscriber(f_sub, d_sub);

    DalSubscriberImpl* handle_ = new DalSubscriberImpl(f_sub, f_listener);
    handle_->SetFastddsDomainParticipant(f_participant);
    d_sub->handle_ = handle_;
    return d_sub;
}

dal_ret_t dal_domain_participant_delete_subscriber(dal_domain_participant_t* participant,
                                                   dal_subscriber_t* sub)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    Subscriber* f_sub = get_fastdds_subscriber(sub);
    if (!f_sub) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds subscriber failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    ReturnCode_t f_ret = f_participant->delete_subscriber(f_sub);
    dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
    if (DAL_RETCODE_OK != d_ret) {
        LOG_ERROR_WITH_LOCATION() << "delete subscriber failed." << d_ret << LOG_ENDL;
        return d_ret;
    }

    EntityMappingRemoveSubscriber(f_sub);

    DalSubscriberImpl* sub_impl = reinterpret_cast<DalSubscriberImpl*>(sub->handle_);
    delete sub_impl;
    dal_subscriber_free(sub);
    reinterpret_cast<dal_subscriber_t*>(sub)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

// DATAREADER
dal_datareader_listener_t*
dal_create_datareader_listener(dal_datareader_listener_callbacks_t callbacks)
{
    dal_datareader_listener_t* listener = dal_datareader_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = (new DataReaderListenerImpl(callbacks));
    if (!listener->handle_) {
        LOG_ERROR() << "DataReaderListenerImpl memory allocation failed." << LOG_ENDL;
        dal_datareader_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_datareader_listener(dal_datareader_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    delete reinterpret_cast<DataReaderListenerImpl*>(listener->handle_);
    dal_datareader_listener_free(listener);
    reinterpret_cast<dal_datareader_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

static bool DalCreateDataReaderParaCheck(dal_subscriber_t* sub, dal_topic_description_t* topic_desc)
{
    if (!sub || !sub->handle_) {
        LOG_ERROR_WITH_LOCATION() << "subscriber is nullptr." << LOG_ENDL;
        return false;
    }

    Subscriber* f_sub = get_fastdds_subscriber(sub);
    if (!f_sub) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds subscriber failed." << LOG_ENDL;
        return false;
    }

    Topic* f_topic = get_fastdds_topic_desc(topic_desc);
    if (!f_topic) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds topic description failed." << LOG_ENDL;
        return false;
    }

    if (!topic_desc->type_support_impl_) {
        LOG_ERROR_WITH_LOCATION() << "topic description type support is nullptr." << LOG_ENDL;
        return false;
    }

    return true;
}

static dal_datareader_t* DalCreateDataReaderByTopicType(dal_subscriber_t* sub,
                                                        dal_topic_description_t* topic_desc,
                                                        const DataReaderQos& f_qos,
                                                        DataReaderListenerImpl* f_listener,
                                                        const StatusMask& f_mask

)
{
    Subscriber* f_sub = get_fastdds_subscriber(sub);
    Topic* f_topic = get_fastdds_topic_desc(topic_desc);
    std::string topic_name = f_topic->get_name();
    dal_datareader_t* d_reader = dal_datareader_allocate();
    if (!d_reader) {
        LOG_ERROR_WITH_LOCATION() << "dal datareader allocate failed." << LOG_ENDL;
        return nullptr;
    }
    d_reader->type_support_impl_ = topic_desc->type_support_impl_;
#ifdef DAL_ENABLE_DMABUFHEAP
    using namespace dal_dmabufheap;
    DmaBufHeapTopicSelector* topic_selector = DmaBufHeapTopicSelector::GetInstance();
    if (topic_selector->IsDbhTopic(topic_name)) {
        DalDataReaderImpl* reader_handle_ = new DalDataReaderImpl(nullptr, nullptr);
        DalSubscriberImpl* sub_handle_ = reinterpret_cast<DalSubscriberImpl*>(sub->handle_);

        if (f_qos.reliability().kind == BEST_EFFORT_RELIABILITY_QOS) {
            reader_handle_->dbh_reader_ = new DalDbhBestEffortReader();
        } else {
            reader_handle_->dbh_reader_ = new DalDbhReliableReader();
        }

        reader_handle_->dbh_reader_->SetFastddsDomainParticipant(
            sub_handle_->FastddsDomainParticipant());
        reader_handle_->dbh_reader_->SetFastddsSubscriber(f_sub);
        reader_handle_->dbh_reader_->SetDbhSendNotifyReaderListener(f_listener);
        reader_handle_->dbh_reader_->SetDbhSendNotifyReaderQos(f_qos);
        reader_handle_->dbh_reader_->SetDbhSendNotifyReaderStatusMask(f_mask);
        reader_handle_->dbh_reader_->SetTypeSupport(d_reader->type_support_impl_);
        reader_handle_->dbh_reader_->SetOriginTopicName(topic_name);
        if (!reader_handle_->dbh_reader_->CreateDbhBuiltinEntities()) {
            LOG_ERROR_WITH_LOCATION() << "create dbh builtin entities failed." << LOG_ENDL;
            delete reader_handle_;
            dal_datareader_free(d_reader);
            return nullptr;
        }

        // d_reader could be created in on_subscription_matched, check it first.
        // EntityMappingAddDataReader(f_reader, d_reader);
        dal_datareader_t* d_reader_tmp = EntityMappingGetOrAddDataReader(
            reader_handle_->dbh_reader_->GetDbhSendNotifyReader(), d_reader);
        if (d_reader_tmp != d_reader) {
            dal_datareader_free(d_reader);
            d_reader_tmp->type_support_impl_ = topic_desc->type_support_impl_;
            d_reader = d_reader_tmp;
        }

        reader_handle_->SetFastddsDomainParticipant(sub_handle_->FastddsDomainParticipant());
        reader_handle_->SetFastddsSubscriber(f_sub);
        reader_handle_->type_ = DalDataReaderImpl::ReaderType::DBH_READER;
        d_reader->handle_ = reader_handle_;
        return d_reader;
    } else if (topic_selector->IsMixedTopic(f_topic->get_name())) {
        LOG_ERROR_WITH_LOCATION() << "mixed topic is not supported yet." << LOG_ENDL;
        dal_datareader_free(d_reader);
        return nullptr;
    } else {
#endif
        auto f_reader = f_sub->create_datareader(f_topic, f_qos, f_listener, f_mask);
        if (!f_reader) {
            LOG_ERROR_WITH_LOCATION() << "create datareader failed." << LOG_ENDL;
            dal_datareader_free(d_reader);
            return nullptr;
        }

        // d_reader could be created in on_subscription_matched, check it first.
        // EntityMappingAddDataReader(f_reader, d_reader);
        dal_datareader_t* d_reader_tmp = EntityMappingGetOrAddDataReader(f_reader, d_reader);
        if (d_reader_tmp != d_reader) {
            dal_datareader_free(d_reader);
            d_reader_tmp->type_support_impl_ = topic_desc->type_support_impl_;
            d_reader = d_reader_tmp;
        }

        DalDataReaderImpl* reader_handle_ = new DalDataReaderImpl(f_reader, f_listener);
        DalSubscriberImpl* sub_handle_ = reinterpret_cast<DalSubscriberImpl*>(sub->handle_);
        reader_handle_->SetFastddsDomainParticipant(sub_handle_->FastddsDomainParticipant());
        reader_handle_->SetFastddsSubscriber(f_sub);
        reader_handle_->SetTypeSupport(d_reader->type_support_impl_);
        reader_handle_->type_ = DalDataReaderImpl::ReaderType::DDS_READER;
        d_reader->handle_ = reader_handle_;
        return d_reader;
#ifdef DAL_ENABLE_DMABUFHEAP
    }
#endif
}

dal_datareader_t* dal_subscriber_create_datareader(dal_subscriber_t* sub,
                                                   dal_topic_description_t* topic_desc,
                                                   dal_datareader_qos_t* qos,
                                                   dal_datareader_listener_t* listener,
                                                   dal_status_mask_t mask)
{
    dal_ret_t d_ret = DAL_RETCODE_OK;

    if (!DalCreateDataReaderParaCheck(sub, topic_desc)) {
        LOG_ERROR_WITH_LOCATION() << "DalCreateDataReaderParaCheck failed." << LOG_ENDL;
        return nullptr;
    }

    DataReaderQos f_qos;
    QosConverter::GetInstance()->ConvertDataReaderQosFromDalToFastdds(qos, &f_qos);
    f_qos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_qos.data_sharing().off();
#endif

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    DataReaderListenerImpl* f_listener =
        listener ? reinterpret_cast<DataReaderListenerImpl*>(listener->handle_) : nullptr;

    return DalCreateDataReaderByTopicType(sub, topic_desc, f_qos, f_listener, f_mask);
}

dal_datareader_t* dal_subscriber_create_datareader_with_profile(dal_subscriber_t* sub,
                                                                dal_topic_description_t* topic_desc,
                                                                dal_datareader_qos_t* qos,
                                                                const char* profile_name,
                                                                dal_datareader_listener_t* listener,
                                                                dal_status_mask_t mask)
{
    if (!profile_name || strlen(profile_name) == 0) {
        return dal_subscriber_create_datareader(sub, topic_desc, qos, listener, mask);
    }

    if (!DalCreateDataReaderParaCheck(sub, topic_desc)) {
        LOG_ERROR_WITH_LOCATION() << "DalCreateDataReaderParaCheck failed." << LOG_ENDL;
        return nullptr;
    }

    DataReaderQos f_qos;
    if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
        try {
            Subscriber* f_sub = get_fastdds_subscriber(sub);
            f_sub->get_datareader_qos_from_profile(profile_name, f_qos);
        }
        catch (...) {
            LOG_ERROR_WITH_LOCATION()
                << "datareader qos profile[" << profile_name << "] in xml are not effective.";
            return nullptr;
        }
    } else {
        QosConverter::GetInstance()->ConvertDataReaderQosFromDalToFastdds(qos, &f_qos);
    }
    f_qos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_qos.data_sharing().off();
#endif

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    DataReaderListenerImpl* f_listener =
        listener ? reinterpret_cast<DataReaderListenerImpl*>(listener->handle_) : nullptr;

    return DalCreateDataReaderByTopicType(sub, topic_desc, f_qos, f_listener, f_mask);
}

static bool DalDeleteDataReaderParaCheck(dal_subscriber_t* sub, dal_datareader_t* reader)
{
    if (!sub || !sub->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param sub invalid." << LOG_ENDL;
        return false;
    }

    if (!reader || !reader->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param reader invalid." << LOG_ENDL;
        return false;
    }

    return true;
}

dal_ret_t dal_subscriber_delete_datareader(dal_subscriber_t* sub, dal_datareader_t* reader)
{
    if (!DalDeleteDataReaderParaCheck(sub, reader)) {
        LOG_ERROR_WITH_LOCATION() << "param check failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    DalDataReaderImpl* reader_impl = reinterpret_cast<DalDataReaderImpl*>(reader->handle_);

#ifdef DAL_ENABLE_DMABUFHEAP
    if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DDS_READER) {
#endif
        Subscriber* f_sub = get_fastdds_subscriber(sub);
        if (!f_sub) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds subscriber failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        DataReader* f_reader = get_fastdds_datareader(reader);
        if (!f_reader) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds datareader failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        ReturnCode_t f_ret = f_sub->delete_datareader(f_reader);
        dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
        if (DAL_RETCODE_OK != d_ret) {
            LOG_ERROR_WITH_LOCATION() << "delete datareader failed." << LOG_ENDL;
            return d_ret;
        }

        EntityMappingRemoveDataReader(f_reader);
#ifdef DAL_ENABLE_DMABUFHEAP
    } else if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DBH_READER) {
        DataReader* f_reader = reader_impl->dbh_reader_->GetDbhSendNotifyReader();
        EntityMappingRemoveDataReader(f_reader);

    } else {
        LOG_ERROR_WITH_LOCATION() << "mixed topic is not supported yet." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }
#endif
    delete reader_impl;
    dal_datareader_free(reader);
    reinterpret_cast<dal_datareader_t*>(reader)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

static bool DalReadAndTakeParaCheck(dal_datareader_t* reader, dal_data data)
{
    if (!reader || !reader->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param reader invalid." << LOG_ENDL;
        return false;
    }

    if (!data) {
        LOG_ERROR_WITH_LOCATION() << "param data invalid." << LOG_ENDL;
        return false;
    }

    return true;
}

dal_ret_t dal_datareader_read_next_sample(dal_datareader_t* reader,
                                          dal_data data,
                                          dal_sample_info_t* d_sample_info)
{
    if (!DalReadAndTakeParaCheck(reader, data)) {
        LOG_ERROR_WITH_LOCATION() << "param check failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    SampleInfoSeq info_seq{1};
    ReturnCode_t f_ret = ReturnCode_t::RETCODE_OK;
    DalDataReaderImpl* reader_impl = reinterpret_cast<DalDataReaderImpl*>(reader->handle_);
#ifdef DAL_ENABLE_DMABUFHEAP
    if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DDS_READER) {
#endif
        DataReader* f_reader = get_fastdds_datareader(reader);
        if (!f_reader) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds datareader failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        f_ret = reader_impl->DDSRead(data, info_seq);
        if (d_sample_info && info_seq.length() == 1) {
            SampleInfoConverter::GetInstance()->ConvertSampleInfoFromFastddsToDal(d_sample_info,
                                                                                  &(info_seq[0]));
        }

        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
#ifdef DAL_ENABLE_DMABUFHEAP
    } else if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DBH_READER) {
        if (!reader_impl->dbh_reader_) {
            LOG_ERROR_WITH_LOCATION() << "dbh_reader is nullptr." << LOG_ENDL;
            return DAL_RETCODE_ERROR;
        }

        LOG_INFO() << "in dmabufheap mode, dal_datareader_read_next_sample is not supported, dal "
                      "call dal_datareader_take_next_sample instead."
                   << LOG_ENDL;
        f_ret = reader_impl->DBHTake(data, info_seq);
        if (d_sample_info && info_seq.length() == 1) {
            SampleInfoConverter::GetInstance()->ConvertSampleInfoFromFastddsToDal(d_sample_info,
                                                                                  &(info_seq[0]));
        }

        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
    } else {
        std::cout << "mixed topic is not supported yet." << std::endl;
    }

    return DAL_RETCODE_ERROR;
#endif
}

dal_ret_t dal_datareader_take_next_sample(dal_datareader_t* reader,
                                          dal_data data,
                                          dal_sample_info_t* d_sample_info)
{
    if (!DalReadAndTakeParaCheck(reader, data)) {
        LOG_ERROR_WITH_LOCATION() << "param check failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    SampleInfoSeq info_seq{1};
    ReturnCode_t f_ret = ReturnCode_t::RETCODE_OK;
    DalDataReaderImpl* reader_impl = reinterpret_cast<DalDataReaderImpl*>(reader->handle_);
#ifdef DAL_ENABLE_DMABUFHEAP
    if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DDS_READER) {
#endif
        DataReader* f_reader = get_fastdds_datareader(reader);
        if (!f_reader) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds datareader failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        f_ret = reader_impl->DDSTake(data, info_seq);
        if (d_sample_info && info_seq.length() == 1) {
            SampleInfoConverter::GetInstance()->ConvertSampleInfoFromFastddsToDal(d_sample_info,
                                                                                  &(info_seq[0]));
        }

        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
#ifdef DAL_ENABLE_DMABUFHEAP
    } else if (reader_impl->type_ == DalDataReaderImpl::ReaderType::DBH_READER) {
        if (!reader_impl->dbh_reader_) {
            LOG_ERROR_WITH_LOCATION() << "dbh_reader is nullptr." << LOG_ENDL;
            return DAL_RETCODE_ERROR;
        }

        f_ret = reader_impl->DBHTake(data, info_seq);
        if (d_sample_info && info_seq.length() == 1) {
            SampleInfoConverter::GetInstance()->ConvertSampleInfoFromFastddsToDal(d_sample_info,
                                                                                  &(info_seq[0]));
        }

        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
    } else {
        std::cout << "mixed topic is not supported yet." << std::endl;
    }
#endif
    return DAL_RETCODE_ERROR;
}

dal_guid_t dal_datareader_get_guid(dal_datareader_t* reader)
{
    dal_guid_t d_guid = {};
    DataReader* f_reader = get_fastdds_datareader(reader);
    if (!f_reader) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds datareader failed." << LOG_ENDL;
        return d_guid;
    }

    auto f_guid = f_reader->guid();
    GuidConverter::GetInstance()->ConvertGuidFromFastddsToDal(&f_guid, &d_guid);
    return d_guid;
}
